#include "base/channel.h"
#include "base/epoll.h"
#include "base/log.h"
#include "base/timeevent.h"
#include "base/timeevery.h"
#include "base/timequeue.h"
#include "base/timeeverymap.h"
#include "base/thread.h"
#include "net/connectpool.h"

#include <sys/time.h>
#include <sys/syscall.h>

#include "base/eventloop.h"
namespace net_tools{

Eventloop::Eventloop(int size,int tid)
  :thread_tid(0),
   wakeup_(false),
   epoll_(new base::Epoll(size,&wakeup_)),
   timequeue_(new base::Timequeue()),
   wakeupfd_(eventfd(0,EFD_NONBLOCK | EFD_CLOEXEC)),
   quit(false),
   mutex_(),
   timeeverymap_(),
   timeeventset_()
{
  wakeup_job_.reserve(_C::DEFAULT_WAKEUP_VECTOR_SIZE);

  base::Channel *channel_ = epoll_->Get_free_channel(wakeupfd_);
  channel_->Set_read_callback([&]{Wakeup_by_other();}); //lambda
  channel_->Enableread();
  epoll_->Add_event(channel_);

  channel_ = epoll_->Get_free_channel(timequeue_->Gettimefd());
  channel_->Set_read_callback([&]{timequeue_->Timeout();}); //lambda
  channel_->Enableread();
  epoll_->Add_event(channel_);
  NT_LOG_TRACE << "EVENTLOOP CREAT" << NT_LOG_ENDL;
};
Eventloop::~Eventloop(){
  close(wakeupfd_);
};
void Eventloop::Loop(){
  if(thread_tid == 0 || base::Thread::thread_tid_ == 0){
    base::Thread::thread_tid_ = syscall(SYS_gettid);
    thread_tid = base::Thread::thread_tid_;
  }
  NT_LOG_INFO << "BEGIN LOOP" << NT_LOG_ENDL;
  while(!quit){
    epoll_->Epoll_wait(_C::EPOLL_FOREVER);
    Wakeup_job();
  }
};
base::Channel* Eventloop::Get_free_channel(int fd){
  return epoll_->Get_free_channel(fd);
};
void Eventloop::Free_channel(base::Channel* channel){
  Run_in_right_thread([&,channel]{channel->~Channel();}); //lambda
};
void Eventloop::Free_channel_right_now(base::Channel* channel){
  epoll_->Free_channel(channel);
};
void Eventloop::Addevent_right(base::Channel* channel){
  Run_in_right_thread([&,channel]{epoll_->Add_event(channel);}); //lambda
};
void Eventloop::Modevent_right(base::Channel* channel){
  Run_in_right_thread([&,channel]{epoll_->Mod_event(channel);}); //lambda
};
void Eventloop::Delevent_right(base::Channel* channel){
  Run_in_right_thread([&,channel]{epoll_->Del_event(channel);}); //lambda
};
void Eventloop::Addevent(base::Channel* channel){
  epoll_->Add_event(channel); //lambda
};
void Eventloop::Modevent(base::Channel* channel){
  epoll_->Mod_event(channel); //lambda
};
void Eventloop::Delevent(base::Channel* channel){
  epoll_->Del_event(channel);; //lambda
};
void Eventloop::Run_in_right_thread(func func_){
  if(Is_in_right_thread()){
    func_();
  }else{
    mutex_.Lock();
    wakeup_job_.push_back(std::move(func_));
    mutex_.Unlock();
    Wakeup();
  }
};
void Eventloop::Run_in_thread(func func_){
  mutex_.Lock();
  wakeup_job_.push_back(std::move(func_));
  mutex_.Unlock();
  Wakeup();
};
base::Timeevent* Eventloop::Run_after(func func_,timeval begin_time,base::Timeeventset* timeeventset){
  timeval nowtime,righttime;
  gettimeofday(&nowtime,NULL);
  righttime.tv_usec = nowtime.tv_usec + begin_time.tv_usec;
  righttime.tv_sec = nowtime.tv_sec + begin_time.tv_sec;
  if(righttime.tv_usec >= 1000000){
    righttime.tv_usec -= 1000000;
    righttime.tv_sec++;
  }
  if(!timeeventset){
    timeeventset = &timeeventset_;
  }
  base::Timeevent* p_timeevent = new base::Timeevent(std::move(func_),righttime,timeeventset);
  Run_in_right_thread([&,p_timeevent]{timequeue_->Addtimeevent(p_timeevent);}); //lambda
  return p_timeevent;
};
base::Timeevent* Eventloop::Run_at(func func_,timeval begin_time,base::Timeeventset* timeeventset){
  if(!timeeventset){
    timeeventset = &timeeventset_;
  }
  base::Timeevent* p_timeevent  = new base::Timeevent(std::move(func_),begin_time,timeeventset);
  Run_in_right_thread([&,p_timeevent]{timequeue_->Addtimeevent(p_timeevent);}); //lambda
  return p_timeevent;
};
base::Timeevery* Eventloop::Run_every(func func_,timeval begin_time,timeval time,int count,const std::string& job_name,base::Timeeverymap* timeeverymap){
  if(!timeeverymap){
    timeeverymap = &timeeverymap_;
  }
  base::Timeevery* timeevery = new base::Timeevery(std::move(func_),begin_time,time,count,this,timeeverymap,job_name);
  Run_in_right_thread([&,timeevery]{base::Channel* channel_ = epoll_->Get_free_channel(timeevery->Get_fd());
                                    timeevery->Init_channel(channel_);
                                    channel_->Set_read_callback([&,timeevery]{timeevery->Time_out();});
                                    channel_->Enableread();
                                    epoll_->Add_event(channel_);}); //lambda
  return timeevery;
};
//only for eventloop self timejob
void Eventloop::Deltimejob_every(const std::string job_name){
  Run_in_right_thread([&,job_name]{timeeverymap_.Del_timejob(job_name);});
};
bool Eventloop::Deltimejob_once(base::Timeevent* timeevent){
  return timeeventset_.Del_timejob(timeevent);
};
void Eventloop::Wakeup_by_other(){
  eventfd_read(wakeupfd_,&wakeup_event_use_);
};
void Eventloop::Wakeup_job(){
  mutex_.Lock();
  if(wakeup_job_.empty()) [[likely]] {
    mutex_.Unlock();
    return;
  }
  std::vector<func> wakeup_doing_job_;
  // wakeup_doing_job_.reserve(_C::DEFAULT_WAKEUP_VECTOR_SIZE);
  wakeup_.store(false);
  wakeup_job_.swap(wakeup_doing_job_);
  mutex_.Unlock();
  int size_ = wakeup_doing_job_.size();
  for(int count = 0;count < size_;++count){
    wakeup_doing_job_[count]();
  }
};
void Eventloop::Wakeup(){
  if(!wakeup_.load()){
    eventfd_write(wakeupfd_,1);
    wakeup_.store(true);
  }
};
void Eventloop::Init_thread_id(){
  thread_tid = syscall(SYS_gettid);
};
bool Eventloop::Is_in_right_thread(){
  return base::Thread::thread_tid_ == thread_tid;
};
std::string* Eventloop::Get_Thread_name(){
  return base::Thread::p_thread_name_;
};

} // namespace net_tools